package com.atguigu.day03;

import com.atguigu.utils.ItemViewCountPerWindow;
import com.atguigu.utils.UserBehavior;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

// 实现全窗口聚合函数
public class Example8 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .readTextFile("/home/zuoyuan/flinktutorial0819/src/main/resources/UserBehavior.csv")
                .map(new MapFunction<String, UserBehavior>() {
                    @Override
                    public UserBehavior map(String value) throws Exception {
                        String[] arr = value.split(",");
                        return new UserBehavior(
                                arr[0], arr[1], arr[2], arr[3],
                                Long.parseLong(arr[4]) * 1000L
                        );
                    }
                })
                .filter(r -> r.type.equals("pv"))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                                .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
                                    @Override
                                    public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                )
                .keyBy(r -> r.itemId)
                .process(new FakeSlidingWindow(60 * 60 * 1000L, 5 * 60 * 1000L))
                .print();

        env.execute();
    }

    public static class FakeSlidingWindow extends KeyedProcessFunction<String, UserBehavior, ItemViewCountPerWindow> {
        private long windowSize; // 窗口长度
        private long windowStep; // 滑动距离

        public FakeSlidingWindow(long windowSize, long windowStep) {
            this.windowSize = windowSize;
            this.windowStep = windowStep;
        }

        // key: 窗口开始时间
        // value：累加器
        private MapState<Long, Long> mapState;

        @Override
        public void open(Configuration parameters) throws Exception {
            mapState = getRuntimeContext().getMapState(
                    new MapStateDescriptor<Long, Long>(
                            "windowStart-elements",
                            Types.LONG,
                            Types.LONG
                    )
            );
        }

        @Override
        public void processElement(UserBehavior value, Context ctx, Collector<ItemViewCountPerWindow> out) throws Exception {
            ArrayList<Long> windowStartTimes = new ArrayList<>();
            long start = getWindowStart(value.ts, windowStep);
            for (long i = start; i > value.ts - windowSize; i -= windowStep) {
                windowStartTimes.add(i);
            }

            for (Long i : windowStartTimes) {
                if (!mapState.contains(i)) {
                    // 如果不包含窗口开始时间对应的窗口
                    mapState.put(i, 1L);
                } else {
                    mapState.put(i, mapState.get(i) + 1L);
                }
            }

            // 注册定时器
            for (long i : windowStartTimes) {
                // 在窗口开始时间+窗口长度-1毫秒处注册定时器
                ctx.timerService().registerEventTimeTimer(i + windowSize - 1L);
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<ItemViewCountPerWindow> out) throws Exception {
            long windowStart = timestamp + 1L - windowSize;
            out.collect(
                    new ItemViewCountPerWindow(
                            ctx.getCurrentKey(),
                            (long) mapState.get(windowStart),
                            windowStart,
                            timestamp + 1L
                    )
            );
            // 删除窗口
            mapState.remove(windowStart);
        }

        // 获取窗口开始时间，例如窗口长度10秒钟，滑动距离5秒钟，事件时间7秒钟
        public static long getWindowStart(long timestamp, long step) {
            return timestamp - (timestamp + step) % step;
        }
    }
}
